package com.feidee.fd.sml.algorithm.component.ml.classification

import com.feidee.fd.sml.algorithm.component.ml.MLParam
import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.classification.LinearSVC

/**
  * @author YongChen，songhaicheng
  * @date 2018/3/21 10:18
  * @description
  * @reviewer songhaicheng
  */
case class LinearSVCParam(
                           override val input_pt: String,
                           override val output_pt: String,
                           override val hive_table: String,
                           override val flow_time: String,
                           override val featuresCol: String,
                           override var labelCol: String,
                           override var predictionCol: String,
                           override val modelPath: String,
                           override val metrics: Array[String],
                           // 原始结果列
                           rawPredictionCol: String,
                           // 正则化系数，>= 0，默认 0.0
                           regParam: Double,
                           // 最大迭代次数，>= 0，默认 100
                           maxIter: Int,
                           // 是否需要计算截距，默认 true
                           fitIntercept: Boolean,
                           // 收敛系数，>= 0，默认 1E-6
                           tol: Double,
                           // 是否对特征进行标准化，默认 true
                           standardization: Boolean,
                           // 样本权重列
                           weightCol: String,
                           // 分类阈值，默认 0.0
                           threshold: Double,
                           // Spark treeAggregate 算子的参数，适度增大可以加快训练速度，>= 2，默认 2
                           aggregationDepth: Int
                         ) extends MLParam {

  def this() = this(null, null, null, null, "features", "label", "prediction", null, new Array[String](0),
    "rawPrediction" , 0.0, 100, true, 1E-6, true, null, 0.0 , 2)

  override def verify(): Unit = {
    super.verify()
    require(maxIter >= 0, "param maxIter can't be negative")
    require(regParam >= 0, "param regParam can't be negative")
    require(tol >= 0, "param tol can't be negative")
    require(aggregationDepth >= 2, "param aggregationDepth must be greater than 1")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("rawPredictionCol" -> rawPredictionCol)
    map += ("regParam" -> regParam)
    map += ("maxIter" -> maxIter)
    map += ("fitIntercept" -> fitIntercept)
    map += ("tol" -> tol)
    map += ("standardization" -> standardization)
    map += ("weightCol" -> weightCol)
    map += ("threshold" -> threshold)
    map += ("aggregationDepth" -> aggregationDepth)
    map
  }
}


class LinearSVCComponent extends  AbstractClassificationComponent[LinearSVCParam] {
  /**
    * 根据参入参数设置算法组件实体
    *
    * @param param 运行参数
    * @return 算法组件
    */
  override def setUp(param: LinearSVCParam): PipelineStage = {
    val lsvc = new LinearSVC()
      .setFeaturesCol(param.featuresCol)
      .setLabelCol(param.labelCol)
      .setPredictionCol(param.predictionCol)
      .setRawPredictionCol(param.rawPredictionCol)
      .setRegParam(param.regParam)
      .setMaxIter(param.maxIter)
      .setFitIntercept(param.fitIntercept)
      .setTol(param.tol)
      .setStandardization(param.standardization)
      .setAggregationDepth(param.aggregationDepth)
      .setThreshold(param.threshold)

    if (tool.isNotNull(param.weightCol)) {
      lsvc.setWeightCol(param.weightCol)
    }

    lsvc

  }
}

object LinearSVCComponent {

  def apply(paramStr: String): Unit = {
    new LinearSVCComponent()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    LinearSVCComponent(args(0))
  }
}

